package com.zsy.flink.source;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.NumberSequenceIterator;

import java.util.Properties;

/**
 * @Author: Zhou ShiYang
 * @Description:
 * @Date:Create：in 2021/8/14 11:23
 **/
public class SourceApp {

    public static void main(String[] args) throws Exception {
        // 创建上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        test02(env);

        env.execute("SourceApp");
    }

    /**
     * 测试接受kafka消息进行处理
     *
     * @param env
     */
    public static void test05(StreamExecutionEnvironment env) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "47.108.87.27:9092");
        properties.setProperty("group.id", "test");
        DataStream<String> stream = env
                .addSource(new FlinkKafkaConsumer<>("flinkTopic", new SimpleStringSchema(), properties));
        System.out.println(stream.getParallelism());
        stream.print();
    }

    public static void test02(StreamExecutionEnvironment env) {
        DataStreamSource<Long> longDataStreamSource = env.fromParallelCollection(new NumberSequenceIterator(1, 10), Long.class).setParallelism(1);
        System.out.println(longDataStreamSource.getParallelism());

        // 默认用全部
        SingleOutputStreamOperator<Long> filter = longDataStreamSource.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long value) throws Exception {
                return value > 5;
            }
        });
        System.out.println(filter.getParallelism());
        filter.print();
    }

    public static void test01(StreamExecutionEnvironment env) {
        //创建上下文的方式
        StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建本地上下文
        StreamExecutionEnvironment.createLocalEnvironment();
        // 设定并行度
        StreamExecutionEnvironment.createLocalEnvironment(3);
        //
        StreamExecutionEnvironment.createLocalEnvironment(new Configuration());
        StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

    }
}
